package org.example.weather.bigscreen.core.component;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.shardingsphere.infra.datanode.DataNode;
import org.example.weather.bigscreen.core.repository.StWeatherUploadMapper;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;

@Slf4j
public class AsyncCreateTableTask implements Callable<Collection<DataNode>> {
    private final Collection<DataNode> dataNodes;
    private final SqlSessionFactory sqlSessionFactory;
    private final RedissonClient redissonClient;
    private String dataSourceTablesCacheKey;
    private String dataSourceName;

    public AsyncCreateTableTask(final Collection<DataNode> dataNodes, final SqlSessionFactory sqlSessionFactory, final RedissonClient redissonClient, final String dataSourceTablesCacheKey, final String dataSourceName) {
        this.dataNodes = dataNodes;
        this.sqlSessionFactory = sqlSessionFactory;
        this.redissonClient = redissonClient;
        this.dataSourceTablesCacheKey = dataSourceTablesCacheKey;
        this.dataSourceName = dataSourceName;
    }

    @Override
    public Collection<DataNode> call() throws Exception {
        if (CollectionUtils.isEmpty(dataNodes)) {
            return dataNodes;
        }
        SqlSession sqlSession = sqlSessionFactory.openSession();
        StWeatherUploadMapper mapper = sqlSession.getMapper(StWeatherUploadMapper.class);
        RSet<String> cacheTableNames = redissonClient.getSet(String.format(dataSourceTablesCacheKey, dataSourceName, mapper.getClass().getSimpleName()));
        List<DataNode> todoDataNodes = this.dataNodes.stream().filter(dataNode -> !cacheTableNames.contains(dataNode.getTableName())).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(todoDataNodes)) {
            mapper.createTablesLike(todoDataNodes);
            cacheTableNames.addAll(todoDataNodes.stream().map(DataNode::getTableName).collect(Collectors.toList()));
            log.info("AsyncCreateTableTask success!");
        }
        return todoDataNodes;
    }
}
